package org.smr.ministore.mqtt.conf;

import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Vertx;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttServerOptions;
import org.smr.ministore.mqtt.comp.MqttProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.nio.charset.Charset;
import java.util.Objects;

@Configuration
public class VertxConfiguration {

    @Bean
    @ConditionalOnMissingClass(value={"io.vertx.core.Vertx"})
    public Vertx initVertx() {
        Vertx vertx = Vertx.vertx();
        return vertx;
    }


    @Bean
    @ConditionalOnMissingClass(value={"io.vertx.mqtt.MqttClient"})
    public MqttClient initVertx(Vertx vertx, MqttServer server) {
        if(Objects.nonNull(server)){

        }
        MqttClient client = MqttClient.create(vertx);
        return client;
    }

    @Bean
    public MqttServer ininMqttServer(Vertx vertx, MqttProperties properties) {

        MqttServerOptions options = new MqttServerOptions()
                .setPort(properties.getPort());
              /*  .setKeyCertOptions(new PemKeyCertOptions()
                        .setKeyPath("./src/test/resources/tls/server-key.pem")
                        .setCertPath("./src/test/resources/tls/server-cert.pem"))
                .setSsl(true);*/

        MqttServer mqttServer = MqttServer.create(vertx, options);
        mqttServer.endpointHandler(endpoint -> {

            // shows main connect info
            System.out.println("MQTT client [" + endpoint.clientIdentifier() + "] request to connect, clean session = " + endpoint.isCleanSession());

            if (endpoint.auth() != null) {
                System.out.println("[username = " + endpoint.auth().getUsername() + ", password = " + endpoint.auth().getPassword() + "]");
            }
            /* if (endpoint.will() != null) {
                System.out.println("[will topic = " + endpoint.will().getWillTopic() + " msg = " + new String(endpoint.will().getWillMessageBytes()) +
                        " QoS = " + endpoint.will().getWillQos() + " isRetain = " + endpoint.will().isWillRetain() + "]");
            }*/
            System.out.println("[keep alive timeout = " + endpoint.keepAliveTimeSeconds() + "]");
            // accept connection from the remote client
            endpoint.accept(false);

            endpoint.publishHandler(message -> {
                System.out.println("Just received message [" + message.payload().toString(Charset.defaultCharset()) + "] with QoS [" + message.qosLevel() + "]");
                System.out.println(message.topicName());
                if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
                    endpoint.publishAcknowledge(message.messageId());
                } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
                    endpoint.publishReceived(message.messageId());
                }
            }).publishReleaseHandler(messageId -> {
                endpoint.publishComplete(messageId);
            }).publishCompletionHandler(messageId -> {
                System.out.println("Received ack for message = " + messageId);
            });
        }).listen(ar -> {
            if (ar.succeeded()) {
                System.out.println("MQTT server is listening on port " + ar.result().actualPort());
            } else {
                System.out.println("Error on starting the server");
                ar.cause().printStackTrace();
            }
        });
        return mqttServer;
    }


}
